package com.j.lemon.learn.pulsar;

import org.apache.pulsar.client.api.*;

/**
 * @Author lijunjun
 * @Date 2020/9/14 11:25
 * @Description
 */
public class ConsumerShared {

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient pulsarClient = PulsarUtils.getClient();

        Consumer<byte[]> consumer = pulsarClient.newConsumer()
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionName("subscriptionType-test")
                .consumerName("a")
                .topic("persistent://4a/audit/subscriptionType-test")
//                .enableRetry(true)
//                .negativeAckRedeliveryDelay(0, TimeUnit.SECONDS)
//                .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(2).deadLetterTopic("persistent://4a/audit/deadLetter").build())
//                .topic("persistent://4a/audit/subscriptionType-test").messageListener(new MessageListener<byte[]>() {
//                    @Override
//                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
//                        String key = msg.getKey();
//                        System.out.println("key:"+key);
//                        String s = new String(msg.getValue());
//                        System.out.println(System.currentTimeMillis()+""+consumer.getConsumerName()+"接收到一条消息："+s);
////                        if(s.startsWith("<88")){
////                            try {
////                                consumer.acknowledge(msg);
////                            } catch (PulsarClientException e) {
////                                e.printStackTrace();
////                            }
////                        }else{
////                            consumer.negativeAcknowledge(msg);
////                        }
//                        consumer.negativeAcknowledge(msg);
//                    }
//                })
                .subscribe();
        while (true){
            Message<byte[]> receive = consumer.receive();
//            consumer.acknowledge(receive);
            System.out.println(new String(receive.getValue()));
        }
    }
}
